package com.wenx.restcamel;

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.DefaultMessage;
import org.apache.camel.model.RouteDefinition;

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.util.List;
import java.util.ArrayList;

public class AppHttp2 {
    public static void main(String[] args) throws Exception {
        CamelContext context = new DefaultCamelContext(); // 1. 创建 CamelContext.
        context.setTracing(true);
        KafkaComponent kafka = new KafkaComponent();
        kafka.setBrokers("10.10.5.212:9092");
        context.addComponent("kafka", kafka);
        context.start(); // 4. 启动CamelContext.
        RouteDefinition rd1 = new RouteDefinition();
        RouteDefinition rd2 = new RouteDefinition();
        rd1.id("001").from("timer:foo?repeatCount=1")
       // rd1.id("001").from("quartz://report?cron=3 * * * * ?&stateful=true")
                .to("http://localhost:8080/demo/hello4?id=123&name=abc")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                      /* Map<String,Object> headerMap = exchange.getIn().getHeaders();
                       String inBody = exchange.getIn().getBody().toString();
                       Map<String,DataHandler> attachmentMap = exchange.getIn().getAttachments();
                       for(Map.Entry<String,Object> eap : headerMap.entrySet()){
                           System.out.println("headerMap: "+eap.getKey()+"="+eap.getValue());
                        }
                        System.out.println("body: "+inBody);
                        for(Map.Entry<String,DataHandler> eap : attachmentMap.entrySet()){
                            System.out.println("attachmentMap: "+eap.getKey()+"="+eap.getValue());
                        }*/
                        DefaultMessage message = (DefaultMessage) exchange.getIn();
                        InputStream bodyStream =  (InputStream)message.getBody();
                        ByteArrayOutputStream result = new ByteArrayOutputStream();
                        byte[] buffer = new byte[1024];
                        int length;
                        while ((length = bodyStream.read(buffer)) != -1) {
                            result.write(buffer, 0, length);
                        }
                        String bodystr = result.toString("UTF-8");
                        StringBuffer sb=new StringBuffer(bodystr);
                        bodystr=sb.append(System.currentTimeMillis()).toString();
                        System.out.println("生产bodystr="+bodystr);
                    }
                })
                .to("kafka:10.10.5.212:9092?topic=test2");
        rd2.id("002").from("kafka:test1?brokers=10.10.5.212:9092"
                + "&maxPollRecords=5000"
                + "&consumersCount=1"
                + "&groupId=test-consumer-group")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        DefaultMessage message = (DefaultMessage) exchange.getIn();
                        String bodystr =  message.getBody().toString();
                        System.out.println("消费bodystr="+bodystr);
                    }
                });
                //.to("stream:out");
        context.addRouteDefinition(rd1);
//        context.addRouteDefinition(rd2);
        /*List<RouteDefinition> li=new ArrayList<RouteDefinition>();
        li.add(rd1);
        li.add(rd2);
        context.addRouteDefinitions(li);*/
        Thread.sleep(Integer.MAX_VALUE);  // 为了保持CamelContext处于工作状态，这里需要sleep主线程
        context.stop(); // 最后停止CamelContext
    }
}
